package com.yanduo.tools

import ch.hsr.geohash.GeoHash
import com.yanduo.utils.JedisPools
import org.apache.commons.lang3.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

/**
  *
  * 用来抽取日志字段中的经纬度，并请求百度api,
  * 获取商圈信息
  *
  * @author Gerry chan
  * 2020/5/17 16:32
  * @version 1.0
  */
object ExtractLatLong2Business {
  def main(args: Array[String]): Unit = {
    //0 校验参数个数
    if (args.length != 1) {
      println(
        """
          |com.yanduo.tools.ExtractLatLong2Business
          |参数：
          |inputPath
        """.stripMargin)
      sys.exit()
    }
    val Array(inputPath) = args

    val sparkConf = new SparkConf()
    sparkConf.setAppName(s"${this.getClass.getSimpleName}")
    sparkConf.setMaster("local[*]")

    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(sparkConf)
    val spark = new SparkSession(sc)

    //解析经纬度,并保存到redis数据库中
    spark.read.parquet(inputPath).select("lat","long")
      .where("lat >3 and lat<54 and long>73 and long<136")
      .distinct()
      .foreachPartition(itr => {
        //jedis客户端获取
        val jedis = JedisPools.getJedis
          itr.foreach(row => {
              val lat = row.getAs[String]("lat").toDouble
              val longs = row.getAs[String]("long").toInt
              val geoHashCode = GeoHash.withCharacterPrecision(lat,longs, 8).toBase32
              //获取经纬度解析后的数据
              val business = BaiduGeoApi.getBusiness(lat+","+longs)
            if (StringUtils.isNotEmpty(business)) {
              //保存到redis数据库中
              jedis.set(geoHashCode, business)
            }
          })
        jedis.close()
    })

    sc.stop()
    spark.stop()
  }

}
